package cn.doitedu.rtdw.etl;

import cn.doitedu.rtdw.beans.SearchAggBean;
import cn.doitedu.rtdw.beans.SearchAggDorisBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/18
 * @Desc: 学大数据，上多易教育
 *   搜索行为olap分析基础表开发
 **/
public class Job05_search_olap_base {
    public static void main(String[] args) throws Exception {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setStateBackend(new HashMapStateBackend()); // 默认的stateBackend
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true)); // 生产中更建议使用rocksdb状态后端



        env.setParallelism(4);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建kafka中dwd行为明细topic的映射表
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "      session_id  string                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,device_type string                                 "+
                        "     ,release_channel string                             "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );


        //tenv.executeSql("select action_time from dwd_events_kafka").print();
        // 聚合，粒度： 一次搜索生命周期的数据聚合成一行
        Table table = tenv.sqlQuery(
                " WITH tmp AS (                                                                               \n " +
                        "     SELECT		                                                                  \n " +
                        "      session_id                                                                     \n " +
                        "     ,event_id                                                                       \n " +
                        "     ,action_time                                                                    \n " +
                        "     ,properties['keyword'] as keyword                                               \n " +
                        "     ,properties['search_id'] as search_id                                           \n " +
                        "     ,cast( properties['res_cnt'] as bigint) as res_cnt                              \n " +
                        "     ,properties['item_seq'] as click_seq                                            \n " +
                        "     ,user_id                                                                        \n " +
                        "     ,province                                                                       \n " +
                        "     ,city                                                                           \n " +
                        "     ,region                                                                         \n " +
                        "     ,page_type                                                                      \n " +
                        "     ,page_service                                                                   \n " +
                        "     ,device_type                                                                    \n " +
                        "     ,release_channel                                                                \n " +
                        "     ,rt                                                                             \n " +
                        "     FROM dwd_events_kafka                                                           \n " +
                        "     WHERE event_id in ('search','search_return','search_click')                     \n " +
                        " )                                                                                   \n " +
                        " SELECT                                                                              \n " +
                        " 	  date_format(to_timestamp_ltz(min(action_time),3),'yyyy-MM-dd') as dt            \n " +
                        " 	  ,user_id                                                                        \n " +
                        " 	  ,session_id                                                                     \n " +
                        "     ,province                                                                       \n " +
                        "     ,city                                                                           \n " +
                        "     ,region                                                                         \n " +
                        "     ,page_type                                                                      \n " +
                        "     ,page_service                                                                   \n " +
                        "     ,device_type                                                                    \n " +
                        "     ,release_channel                                                                \n " +
                        "     ,search_id	                                                                  \n " +
                        " 	,keyword                                                                          \n " +
                        " 	,min(action_time)  as  search_time -- 发起时间                                     \n " +
                        " 	,max(res_cnt)      as  res_cnt     -- 返回条数                                     \n " +
                        " 	,count(1) filter(where event_id = 'search_click' )  as  click_cnt   -- 点击次数    \n " +
                        " FROM TABLE(                                                                         \n " +
                        "     TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE)                            \n " +
                        " )                                                                                   \n " +
                        " GROUP BY                                                                            \n " +
                        "     window_start                                                                    \n " +
                        " 	  ,window_end                                                                     \n " +
                        " 	  ,user_id                                                                        \n " +
                        " 	  ,session_id                                                                     \n " +
                        "     ,province                                                                       \n " +
                        "     ,city                                                                           \n " +
                        "     ,region                                                                         \n " +
                        "     ,page_type                                                                      \n " +
                        "     ,page_service                                                                   \n " +
                        "     ,device_type                                                                    \n " +
                        "     ,release_channel                                                                \n " +
                        "     ,search_id                                                                      \n " +
                        "     ,keyword	                                                                      \n "
        );


        // 请求http接口，获取搜索词的近义词和分词
        // 表转流
        DataStream<SearchAggBean> dataStream = tenv.toDataStream(table, SearchAggBean.class);
        SingleOutputStreamOperator<SearchAggDorisBean> resultStream = dataStream.keyBy(SearchAggBean::getKeyword)
                .process(new KeyedProcessFunction<String, SearchAggBean, SearchAggDorisBean>() {

                    CloseableHttpClient client;
                    SearchAggDorisBean searchAggDorisBean;
                    HttpPost post;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        client = HttpClientBuilder.create().build();
                        post = new HttpPost("http://doitedu:8081/api/post/simwords");
                        post.addHeader("Content-type", "application/json; charset=utf-8");
                        post.addHeader("Accept", "application/json");

                        searchAggDorisBean = new SearchAggDorisBean();
                    }

                    @Override
                    public void processElement(SearchAggBean bean, KeyedProcessFunction<String, SearchAggBean, SearchAggDorisBean>.Context ctx, Collector<SearchAggDorisBean> out) throws Exception {

                        // 提取bean中搜索词
                        String keyword = bean.getKeyword();

                        // 请求接口

                        post.setEntity(new StringEntity("{\"origin\":\" " + keyword + "\"}","utf-8"));
                        CloseableHttpResponse response = client.execute(post);
                        HttpEntity entity = response.getEntity();
                        String responseJson = EntityUtils.toString(entity);

                        // 解析结果，提取近义词和分词
                        JSONObject jsonObject = JSON.parseObject(responseJson);
                        String split_keyword = jsonObject.getString("words");
                        String similar_keyword = jsonObject.getString("similarWord");


                        // 补全输出信息
                        searchAggDorisBean.set(bean, split_keyword, similar_keyword);

                        // 输出
                        out.collect(searchAggDorisBean);

                    }
                });


        resultStream.print();


        // 流转表
        tenv.createTemporaryView("res",resultStream);
        // 创建doris物理表的映射表

        // 最后写一个insert ... into



        env.execute();

    }
}
